﻿using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using PServiceBus.Services.Gateway.Runtime;
using PServiceBus.Transports;
using System.Diagnostics;
using System.Threading;
using PServiceBus.Core.Runtime.Transports;
using PServiceBus.Core.Interface;

namespace StressTest {
    class Program {
        static readonly int MAX_MESSAGES = 1000000;
        static readonly int BATCH_SIZE = 1000;
        static readonly int SIZE = (int)Math.Floor(MAX_MESSAGES / (decimal)BATCH_SIZE);
        static readonly int BATCH_SIZE_DIFF = MAX_MESSAGES - ((SIZE - 1) * BATCH_SIZE);
        static int _counter = 0;
        static readonly bool USE_RABBIT_MQ = false;

        static void PublishMessageInBatchOfThousand() {
            var collection = new List<List<StressRequest>>();
            var sw = new Stopwatch();
            var msg = StressRequest.New;
            sw.Start();
            for (var i = 0; i < SIZE; i++) {
                var count = (i + 1) * BATCH_SIZE;
                count = count > MAX_MESSAGES ? BATCH_SIZE_DIFF : BATCH_SIZE;
                var list = new List<StressRequest>();
                for (var j = 0; j < count; j++) list.Add(msg);
                Topic.PublishMessages(list);
            }
            sw.Stop();
            Console.WriteLine("Published Message to Bus in {0} Millisecond(s)", sw.ElapsedMilliseconds);
        }

        static void Main(string[] args) {
            ESB.Authenticate("rpgmaker", "password1");

            if (!Topic.Exists<StressRequest>())
                Topic.Register<StressRequest>();

            ITransport transport = null;

            if (USE_RABBIT_MQ) {
                transport = Transport.New<RabbitMQTransport>(t => {
                    t.Format = TransportFormat.Xml;
                    t.Path = "localhost:5672;userID=guest;password=guest;queue=stressqueue";
                });
            } else {
                transport = PServiceBus.MessageProxy.TcpHelper.Transport;
                transport.Format = TransportFormat.Xml;
                (transport as TcpTransport).Port = PServiceBus.MessageProxy.TcpHelper.GetAvailablePort();
            }

            var subscriber =
                Subscriber.New("StressTestSubscriber")
                .SubscribeTo(Topic.Select<StressRequest>())
                .AddTransportFor<StressRequest>("stressTransport", transport);

            subscriber.Save();
            Stopwatch sw = null;

            subscriber.OnMessageReceived<StressRequest>("stressTransport",
                errorAction: error =>
                {
                    error.Continue = true;
                    Console.WriteLine(error.Error.ToString());
                },
                interval: TimeSpan.FromMilliseconds(1),
                batchSize: BATCH_SIZE, handleInListenerThread: false, 
                actionList: msgs => {
                    if (sw == null) {
                        sw = new Stopwatch();
                        sw.Start();
                    }
                    var count = _counter + msgs.Count;
                    Interlocked.Exchange(ref _counter, count);
                    if (_counter >= MAX_MESSAGES) {
                        sw.Stop();
                        Console.WriteLine("Finished processing(publishing/receiving) {0} in {1} Millisecond(s)",
                            _counter, sw.ElapsedMilliseconds);
                    }
                });

            Console.WriteLine("Sending messages and will receive response soon...");

            PublishMessageInBatchOfThousand();


            Console.ReadLine();
            subscriber.Delete();
            
            TransportHandlers.ShutDown();
        }
    }
}
